package com.lagou;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.IntegerDeserializer;

import java.util.Collections;
import java.util.Properties;

public class ConsumerDemo {

    public static void main(String[] args) {

        // 属性对象
        Properties properties = new Properties();
        // 指定集群节点
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"172.16.167.20:9092,172.16.167.20:9093,172.16.167.20:9094");
        // 反序列化类
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,IntegerDeserializer.class);
        // 指定分组名称
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"lagou_grop1");
        // 消息消费者对象
        KafkaConsumer<String,Integer> consumer = new KafkaConsumer<String, Integer>(properties);
        // 主题
        String topic = "lagou";
        // 订阅消息
        consumer.subscribe(Collections.singletonList(topic));
        while (true) {
            // 获取消息的方法是一个阻断式方法
            ConsumerRecords<String,Integer> records = consumer.poll(500);
            for (ConsumerRecord<String,Integer> record : records) {
                System.out.println("主题：" + record.topic() + "，偏移量：" + record.offset() + "，msg:" + record.value());
            }
        }

    }

}
